#!/usr/bin/env python3
import socketserver
import base64
import time
import sqlite3
import logging
from threading import Lock, Timer, Thread
import socket
import os
import sys
import threading
import datetime
import signal
from flask import Flask, request, render_template, redirect, url_for, session, send_from_directory
import psutil


# ===================================
# 全局配置与常量
# ===================================
DEBUG = True
VERSION = '1.9.9'
CONTACT_EMAIL = 'i@jia.by'
HOST = '0.0.0.0'
NTRIP_PORT = 2101
WEB_PORT = 5757
BUFFER_MAXLEN = 2000
ALIPAY_QR_URL = 'https://2rtk.rampump.cn/alipay.jpg'
WECHAT_QR_URL = 'https://2rtk.rampump.cn/wechat.jpg'
BROADCAST_INTERVAL = 0.01  # 广播线程间隔（10ms）
CLIENT_TIMEOUT = 30  # 客户端超时时间（秒）


# ===================================
# 线程安全的RingBuffer实现
# ===================================
class RingBuffer:
    """线程安全的环形缓冲区，用于高效管理RTCM数据缓存"""
    def __init__(self, maxlen):
        self.maxlen = maxlen
        self.buffer = []
        self.lock = Lock()  # 确保多线程安全

    def append(self, item):
        """添加数据项，超出长度时自动移除 oldest 数据"""
        with self.lock:
            if len(self.buffer) >= self.maxlen:
                self.buffer.pop(0)  # 移除最旧数据
            self.buffer.append(item)

    def get_since(self, timestamp):
        """获取指定时间戳之后的所有数据（线程安全）"""
        with self.lock:
            # 筛选出时间戳大于指定值的数据
            return [(ts, chunk) for ts, _, chunk in self.buffer if ts > timestamp]

    def __len__(self):
        """获取当前缓冲区长度"""
        with self.lock:
            return len(self.buffer)


# ===================================
# 全局变量与锁
# ===================================
# 数据缓存与锁（使用RingBuffer替代deque）
rtcm_lock = Lock()
rtcm_buffers = {}  # {mount: RingBuffer}
clients_lock = Lock()
authenticated_clients = []  # 元素: {socket, user, mount, agent, addr, last_sent_ts, last_active, protocol_version}

# 线程控制
broadcast_thread_running = True
START_TIME = time.time()

# 日志配置
logging.basicConfig(
    level=logging.DEBUG if DEBUG else logging.WARNING,
    format='[%(asctime)s] %(levelname)s %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)
logger = logging.getLogger('2RTK')

# 横幅信息
BANNER = r"""
    ██████╗ ██████╗ ████████╗██╗  ██╗
    ╚════██╗██╔══██╗╚══██╔══╝██║ ██╔╝
     █████╔╝██████╔╝   ██║   █████╔╝
    ██╔═══╝ ██╔══██╗   ██║   ██  ██
    ███████╗██║  ██║   ██║   ██╗  ██╗
    ╚══════╝╚═╝  ╔═╝   ╚═╝   ╚═╝  ╚═╝
         2RTK Ntrip Caster/{}
""".format(VERSION)


# ===================================
# 工具函数
# ===================================
def dbg(*args, **kwargs):
    """调试输出（仅DEBUG模式生效）"""
    if DEBUG:
        print(*args, **kwargs)


def format_time_duration(seconds):
    """格式化时间为天/时/分/秒"""
    days = int(seconds // 86400)
    seconds %= 86400
    hours = int(seconds // 3600)
    seconds %= 3600
    minutes = int(seconds // 60)
    secs = int(seconds % 60)

    parts = []
    if days > 0:
        parts.append(f"{days} 天")
    if hours > 0 or parts:
        parts.append(f"{hours} 小时")
    if minutes > 0 or parts:
        parts.append(f"{minutes} 分钟")
    parts.append(f"{secs} 秒")
    return " ".join(parts)


# ===================================
# 数据库初始化与管理
# ===================================
def init_db():
    """初始化SQLite数据库表结构"""
    conn = sqlite3.connect('2rtk.db')
    c = conn.cursor()

    # 管理员表
    c.execute('''
    CREATE TABLE IF NOT EXISTS admins (
        id INTEGER PRIMARY KEY,
        username TEXT NOT NULL UNIQUE,
        password TEXT NOT NULL
    )
    ''')
    
    # 用户表（NTRIP客户端用户）
    c.execute('''
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY,
        username TEXT NOT NULL UNIQUE,
        password TEXT NOT NULL
    )
    ''')
    
    # 挂载点表
    c.execute('''
    CREATE TABLE IF NOT EXISTS mounts (
        id INTEGER PRIMARY KEY,
        mount TEXT NOT NULL UNIQUE,
        password TEXT NOT NULL
    )
    ''')
    
    # 运行中挂载点表
    c.execute('''
    CREATE TABLE IF NOT EXISTS running_mounts (
        id INTEGER PRIMARY KEY,
        mount TEXT NOT NULL,
        start_time INTEGER NOT NULL,
        identifier TEXT,
        format TEXT,
        format_details TEXT,
        carrier INTEGER DEFAULT 0,
        nav_system TEXT,
        network TEXT,
        country TEXT,
        latitude REAL,
        longitude REAL,
        nmea INTEGER DEFAULT 0,
        solution INTEGER DEFAULT 0,
        generator TEXT,
        compression TEXT,
        authentication TEXT,
        fee TEXT DEFAULT 'N',
        bitrate INTEGER,
        misc TEXT,
        update_time INTEGER,
        UNIQUE (mount),
        FOREIGN KEY (mount) REFERENCES mounts(mount)
            ON DELETE CASCADE
            ON UPDATE CASCADE
    )
    ''')
    
    # 服务启动时重置运行中挂载点
    c.execute("DELETE FROM running_mounts")
    conn.commit()
    print("运行中的挂载点列表已清空")
    
    # 创建默认管理员（首次启动）
    c.execute("SELECT * FROM admins")
    if not c.fetchone():
        c.execute("INSERT INTO admins (username, password) VALUES ('admin', 'admin')")
        print("已创建默认管理员: admin/admin（请首次登录后修改）")
    
    conn.commit()
    conn.close()
    print("数据库初始化完成")


# ===================================
# 控制台横幅与状态刷新
# ===================================
def clear_banner():
    """定时刷新控制台横幅与状态信息"""
    os.system('cls' if os.name == 'nt' else 'clear')
    print(BANNER)
    
    # 显示已连接用户数
    with clients_lock:
        connected_user_count = len(authenticated_clients)
    print(f"已连接的用户数: {connected_user_count}")
    
    # 显示运行中挂载点
    conn = sqlite3.connect('2rtk.db')
    c = conn.cursor()
    c.execute("SELECT mount FROM running_mounts")
    online_mounts = [row[0] for row in c.fetchall()]
    conn.close()
    print(f"运行中挂载点: {', '.join(online_mounts)}")
    
    # 100秒后再次刷新
    t = Timer(100, clear_banner)
    t.daemon = True
    t.start()


# ===================================
# 全局广播线程（优化版）
# ===================================
class BroadcastThread(Thread):
    """独立广播线程：使用RingBuffer和集中广播架构"""
    def run(self):
        global broadcast_thread_running
        logger.info("广播线程已启动（RingBuffer+集中广播架构）")
        while broadcast_thread_running:
            try:
                self._broadcast_all()  # 广播所有挂载点数据
                self._cleanup_timeout_clients()  # 清理超时客户端
                time.sleep(BROADCAST_INTERVAL)  # 控制CPU负载
            except Exception as e:
                logger.error(f"广播线程异常: {e}")
                time.sleep(1)  # 异常时延迟重试
        logger.info("广播线程已停止")

    def _broadcast_all(self):
        """遍历所有挂载点，向订阅的客户端广播数据（集中广播架构）"""
        with rtcm_lock:
            # 复制挂载点快照，减少锁占用时间
            current_mounts = list(rtcm_buffers.items())
        
        for mount, buffer in current_mounts:
            # 获取订阅该挂载点的客户端（复制列表避免锁定冲突）
            with clients_lock:
                clients = [c for c in authenticated_clients if c['mount'] == mount]
            
            # 向每个客户端发送未接收的数据
            for client in clients:
                self._send_to_client(client, buffer)

    def _send_to_client(self, client, buffer):
        """向单个客户端发送缓存中的新数据（基于RingBuffer优化）"""
        try:
            # 从RingBuffer获取客户端未接收的数据（高效筛选）
            new_data = buffer.get_since(client['last_sent_ts'])
            if not new_data:
                return  # 无新数据则跳过
            
            # 发送数据（区分NTRIP版本）
            for ts, chunk in new_data:
                if client['protocol_version'] == 'NTRIP v2.0':
                    # v2.0分块编码（符合规范）
                    chunk_data = f"{len(chunk):X}\r\n".encode() + chunk + b"\r\n"
                    client['socket'].sendall(chunk_data)
                else:
                    # v1.0直接发送原始数据
                    client['socket'].sendall(chunk)
                
                # 更新客户端状态（跟踪最后发送的时间戳和活动时间）
                client['last_sent_ts'] = ts
                client['last_active'] = time.time()

        except BlockingIOError:
            # 网络拥堵时的临时错误，不踢客户端（仅记录调试信息）
            logger.debug(f"客户端暂时无法接收数据 (addr={client['addr']}, mount={client['mount']})")
        except Exception as e:
            # 确认真实连接错误时才移除客户端
            logger.warning(f"向客户端发送数据失败 (addr={client['addr']}, mount={client['mount']}): {e}")
            self._remove_client(client)

    def _cleanup_timeout_clients(self):
        """清理超过超时时间的客户端"""
        now = time.time()
        with clients_lock:
            timeout_clients = [c for c in authenticated_clients if now - c['last_active'] > CLIENT_TIMEOUT]
        
        for client in timeout_clients:
            logger.info(f"客户端超时 (addr={client['addr']}, mount={client['mount']}), 已清理")
            self._remove_client(client)

    def _remove_client(self, client):
        """安全移除客户端并关闭socket"""
        # 关闭socket（先shutdown再close，确保资源释放）
        try:
            client['socket'].shutdown(socket.SHUT_RDWR)
            client['socket'].close()
        except Exception as e:
            logger.debug(f"关闭客户端socket失败: {e}")
        
        # 从客户端列表中移除
        with clients_lock:
            if client in authenticated_clients:
                authenticated_clients.remove(client)


# ===================================
# NTRIP服务器处理类
# ===================================
class Handler(socketserver.BaseRequestHandler):
    """处理NTRIP客户端连接（SOURCE上传/GET下载）"""
    
    def handle(self):
        """主处理入口：解析请求并分发到对应逻辑"""
        dbg(f"新连接: {self.client_address}")
        try:
            raw = self.request.recv(1024).decode(errors='ignore')
        except Exception as e:
            logger.error(f"接收数据失败: {e}")
            return
        if not raw:
            return
        
        req, hdrs = self._parse(raw)
        protocol_version = self.get_protocol_version(req, hdrs)
        
        if req.startswith('SOURCE'):
            self._source(req, hdrs, protocol_version)  # 处理挂载点数据上传
        elif req.startswith('GET'):
            self._get(req, hdrs, protocol_version)      # 处理客户端数据下载
        else:
            logger.warning(f"未知请求: {req}")
            self._send_error_response(protocol_version, 400)  # 未知请求返回400


    def get_protocol_version(self, req, hdrs):
        """解析NTRIP协议版本（v1.0/v2.0）"""
        parts = req.split()
        if len(parts) >= 3:
            protocol = parts[-1]
            # NTRIP v2.0需同时满足HTTP/1.1和Ntrip-Version头
            if protocol == 'HTTP/1.1' and hdrs.get('Ntrip-Version') == 'Ntrip/2.0':
                return 'NTRIP v2.0'
            # NTRIP v1.0使用HTTP/1.0或无Ntrip-Version头的HTTP/1.1
            elif protocol == 'HTTP/1.0' or (protocol == 'HTTP/1.1' and 'Ntrip-Version' not in hdrs):
                return 'NTRIP v1.0'
        return 'N/A'


    def _parse(self, raw: str):
        """解析HTTP请求行与头部"""
        lines = raw.splitlines()
        req = lines[0] if lines else ''
        hdrs = {}
        for l in lines[1:]:
            if ': ' in l:
                k, v = l.split(': ', 1)
                hdrs[k.strip()] = v.strip()
        return req, hdrs


    def get_client_model(self, agent):
        """从User-Agent中提取客户端型号"""
        if agent and 'NTRIP' in agent:
            return agent.split('NTRIP', 1)[1].strip()
        return 'N/A'


    # ===================================
    # 响应工具函数
    # ===================================
    def _send_error_response(self, protocol_version, status_code):
        """发送错误响应（根据协议版本适配格式）"""
        if protocol_version == 'NTRIP v2.0':
            responses = {
                400: b'HTTP/1.1 400 Bad Request\r\n\r\n',
                401: b'HTTP/1.1 401 Unauthorized\r\n\r\n',
                404: b'HTTP/1.1 404 Not Found\r\n\r\n',
                500: b'HTTP/1.1 500 Internal Server Error\r\n\r\n'
            }
        else:  # NTRIP v1.0 或未知版本
            responses = {
                400: b'ICY 400 Bad Request\r\n',
                401: b'ICY 401 Unauthorized\r\n',
                404: b'ICY 404 Not Found\r\n',
                500: b'ICY 500 Internal Server Error\r\n'
            }
        self.request.sendall(responses.get(status_code, responses[500]))


    def _challenge(self, protocol_version):
        """发送认证挑战（要求客户端提供账号密码）"""
        if protocol_version == 'NTRIP v2.0':
            self.request.sendall(b'HTTP/1.1 401 Unauthorized\r\nWWW-Authenticate: Basic realm="NTRIP"\r\nContent-Length: 0\r\n\r\n')
        else:
            self.request.sendall(b'ICY 401 Unauthorized\r\nWWW-Authenticate: Basic realm="NTRIP"\r\n\r\n')


    def _send_auth_failure(self, protocol_version):
        """发送认证失败响应"""
        if protocol_version == 'NTRIP v2.0':
            self.request.sendall(b'HTTP/1.1 401 Unauthorized\r\n\r\n')
        else:
            self.request.sendall(b'ICY 401 Unauthorized\r\n')


    # ===================================
    # SOURCE请求处理（挂载点数据上传）
    # ===================================
    def _source(self, req: str, hdrs: dict, protocol_version):
        """处理SOURCE请求（基准站数据上传）"""
        dbg("SOURCE上传请求验证")
        p = req.strip().split()

        # 解析请求格式（标准/非标准）
        mount = None
        password = None
        auth_mode = None  # 'standard'（HTTP认证）/'legacy'（RTKLIB模式）

        if len(p) >= 3:
            # 标准格式：SOURCE /mount HTTP/1.0/1.1
            if p[0].upper() == 'SOURCE' and p[1].startswith('/') and p[2].startswith('HTTP/'):
                auth_mode = 'standard'
                mount = p[1].lstrip('/')
            # 非标准格式（RTKLIB）：SOURCE <password> <mount>
            elif p[0].upper() == 'SOURCE' and not p[2].startswith('HTTP/'):
                auth_mode = 'legacy'
                password = p[1]
                mount = p[2]

        # 格式解析失败
        if not auth_mode or not mount or (not password and auth_mode != 'standard'):
            self._send_error_response(protocol_version, 400)
            logger.warning(f"无法解析上传请求格式: {req}")
            return

        # 标准模式：从Authorization头解析密码
        if auth_mode == 'standard':
            auth_header = hdrs.get('Authorization')
            if not auth_header or not auth_header.startswith('Basic '):
                self._challenge(protocol_version)
                return
            try:
                b64data = auth_header.split()[1]
                username, password = base64.b64decode(b64data).decode().split(':', 1)
            except Exception as e:
                self._send_auth_failure(protocol_version)
                logger.warning(f"Base64解析失败: {e}")
                return

        # 验证挂载点密码
        conn = sqlite3.connect('2rtk.db')
        c = conn.cursor()
        c.execute("SELECT password FROM mounts WHERE mount = ?", (mount,))
        result = c.fetchone()
        if not result or result[0] != password:
            conn.close()
            self._send_auth_failure(protocol_version)
            dbg(f"上传认证失败，密码错误: mount={mount}, addr={self.client_address}")
            return

        # 发送成功响应
        if protocol_version == 'NTRIP v2.0':
            self.request.sendall(b'HTTP/1.1 200 OK\r\nNtrip-Version: Ntrip/2.0\r\n\r\n')
        else:
            self.request.sendall(b'ICY 200 OK\r\n')

        # 记录日志
        agent = self.get_client_model(hdrs.get('Source-Agent', ''))
        logger.info(f"上传认证成功: mount={mount}, agent={agent}, addr={self.client_address}, 模式={auth_mode}")
        print(f"⟳ {mount} 上传认证成功 来自{self.client_address}, 基准站:{agent}, 模式={auth_mode}")

        # 更新运行中挂载点记录
        current_time = time.time()
        try:
            c.execute('''INSERT OR REPLACE INTO running_mounts (
                mount, start_time, identifier, format, format_details,
                carrier, nav_system, network, country, latitude,
                longitude, nmea, solution, generator, compression,
                authentication, fee, bitrate, misc, update_time
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)''', (
                mount, current_time, 'none', 'RTCM 3.3', '1005(10)',
                0, 'GPS', '2RTK', 'CHN', 0.0, 0.0, 0, 0, agent,
                'none', 'B', 'N', 500, 'NO', current_time
            ))
            conn.commit()
        except Exception as e:
            logger.error(f"挂载点数据库记录失败: {e}")
        conn.close()

        # 生成挂载点列表
        self._generate_mount_list()
        print(f"⟳ {mount} 运行中挂载点已更新")

        # 初始化RTCM缓存（使用RingBuffer替代deque）
        with rtcm_lock:
            if mount not in rtcm_buffers:
                rtcm_buffers[mount] = RingBuffer(BUFFER_MAXLEN)

        # 接收并缓存RTCM数据（广播由独立线程处理）
        try:
            self.request.setblocking(1)  # 上传端用阻塞模式接收
            while True:
                chunk = self.request.recv(4096)
                if not chunk:
                    break  # 连接关闭
                ts = time.time()
                with rtcm_lock:
                    if mount in rtcm_buffers:  # 确保挂载点仍存在
                        rtcm_buffers[mount].append((ts, mount, chunk))
                # 每10秒更新挂载点最后更新时间
                if int(ts) % 10 == 0:
                    conn = sqlite3.connect('2rtk.db')
                    c = conn.cursor()
                    c.execute("UPDATE running_mounts SET update_time = ? WHERE mount = ?", (ts, mount))
                    conn.commit()
                    conn.close()
        except Exception as e:
            logger.warning(f"基站上传断开: {e}")

        # 清理：移除运行中挂载点记录
        conn = sqlite3.connect('2rtk.db')
        c = conn.cursor()
        c.execute("DELETE FROM running_mounts WHERE mount = ?", (mount,))
        conn.commit()
        conn.close()

        # 清理RTCM缓存
        with rtcm_lock:
            if mount in rtcm_buffers:
                del rtcm_buffers[mount]

        logger.info(f"上传连接断开: {mount}, addr={self.client_address}")
        self._generate_mount_list()


    # ===================================
    # GET请求处理（客户端数据下载）
    # ===================================
    def _get(self, req: str, hdrs: dict, protocol_version):
        """处理GET请求（客户端下载RTCM数据）"""
        dbg("用户端请求验证中")
        parts = req.split()
        if len(parts) < 3:
            return
        path = parts[1]
        mount = path.lstrip('/')

        # 请求挂载点列表（GET /）
        if path == '/':
            logger.info(f"请求挂载点列表: {self.client_address}")
            agent = self.get_client_model(hdrs.get('User-Agent', ''))
            print(f"请求挂载点列表，客户端: {agent}, 来自：{self.client_address}")

            # 读取挂载点列表
            try:
                with open('mount_list.txt', 'r') as f:
                    tbl = f.read()
            except FileNotFoundError:
                tbl = 'ENDSOURCETABLE;'

            # 构建响应
            date = datetime.datetime.now(datetime.timezone.utc).strftime('%a, %d %b %Y %H:%M:%S GMT')
            if protocol_version == 'NTRIP v2.0':
                resp = '\r\n'.join([
                    'HTTP/1.1 200 OK',
                    'Ntrip-Version: Ntrip/2.0',
                    f'Server: 2RTK Caster/{VERSION}',
                    f'Date: {date}',
                    'Content-Type: text/plain',
                    f'Content-Length: {len(tbl)}',
                    'Connection: close', '', tbl
                ])
            else:
                resp = '\r\n'.join([
                    'SOURCETABLE 200 OK',
                    f'Server: 2RTK Caster/{VERSION}',
                    f'Date: {date}',
                    'Content-Type: text/plain',
                    f'Content-Length: {len(tbl)}',
                    'Connection: close', '', tbl
                ])

            self.request.sendall(resp.encode())
            print(f"已向{self.client_address}, {agent}发送挂载点列表")
            return

        # 请求具体挂载点数据
        agent = self.get_client_model(hdrs.get('User-Agent', ''))
        logger.info(f"请求挂载点 {mount} 下载认证，客户端: {agent}, 来自：{self.client_address}, 协议版本: {protocol_version}")
        print(f"请求挂载点 {mount} 下载认证，客户端: {agent}, 来自：{self.client_address}, 协议版本: {protocol_version}")

        # 验证挂载点是否存在
        conn = sqlite3.connect('2rtk.db')
        c = conn.cursor()
        c.execute("SELECT mount FROM mounts WHERE mount =?", (mount,))
        if not c.fetchone():
            self._send_error_response(protocol_version, 404)
            conn.close()
            return

        # 验证用户认证
        auth = hdrs.get('Authorization')
        if not auth:
            conn.close()
            return self._challenge(protocol_version)

        # 解析并验证用户密码
        try:
            m, cred = auth.split(' ', 1)
            if m.upper() != 'BASIC':
                raise ValueError("非Basic认证")
            u, pw = base64.b64decode(cred).decode().split(':', 1)
            c.execute("SELECT password FROM users WHERE username =?", (u,))
            result = c.fetchone()
            if not result or result[0] != pw:
                raise ValueError("密码错误")
        except:
            self._send_auth_failure(protocol_version)
            conn.close()
            return

        # 记录认证成功的客户端（使用last_sent_ts跟踪发送进度）
        now = time.time()
        with clients_lock:
            # 限制同用户同挂载点3个并发连接
            sess = [c for c in authenticated_clients if c['user'] == u and c['mount'] == mount]
            if len(sess) >= 3:
                old = min(sess, key=lambda x: x['last_sent_ts'])
                self._remove_client(old)  # 移除旧连接
            
            # 添加新客户端（使用last_sent_ts替代auth_time跟踪数据发送）
            client = {
                'socket': self.request,
                'user': u,
                'mount': mount,
                'agent': agent,
                'addr': self.client_address,
                'last_sent_ts': now - 5,  # 允许接收5秒内历史数据
                'last_active': now,        # 初始化最后活动时间
                'protocol_version': protocol_version
            }
            authenticated_clients.append(client)

        logger.info(f"下载认证通过，用户: {u}来自{self.client_address}使用客户端: {agent}, 协议 {protocol_version}至挂载点 {mount}")
        print(f"下载认证通过，用户: {u}来自{self.client_address}使用客户端: {agent}, 协议 {protocol_version}挂载点 {mount}")

        # 发送成功响应
        date = datetime.datetime.now(datetime.timezone.utc).strftime('%a, %d %b %Y %H:%M:%S GMT')
        if protocol_version == 'NTRIP v2.0':
            hdr = '\r\n'.join([
                'HTTP/1.1 200 OK',
                'Ntrip-Version: Ntrip/2.0',
                f'Server: 2RTK Caster/{VERSION}',
                f'Date: {date}',
                'Cache-Control: no-store,max-age=0',
                'Pragma: no-cache',
                'Connection: close',
                'Content-Type: gnss/data',
                'Transfer-Encoding: chunked',
                ''
            ])
        else:
            hdr = 'ICY 200 OK\r\n'
        self.request.sendall(hdr.encode())
        print(f"开始从 {mount} 向{self.client_address}{agent}用户: {u}发送数据流...")

        # 客户端保持连接（数据发送由广播线程处理）
        try:
            self.request.setblocking(0)  # 非阻塞模式避免线程阻塞
            # 等待连接关闭（通过广播线程检测）
            while True:
                time.sleep(1)
                with clients_lock:
                    if client not in authenticated_clients:
                        break
        except Exception as e:
            logger.debug(f"客户端连接中断: {e}")
        finally:
            # 确保客户端被移除
            with clients_lock:
                if client in authenticated_clients:
                    self._remove_client(client)
            conn.close()
            logger.info(f"下载连接关闭: {self.client_address}用户: {u}已从挂载点 {mount} 断开")
            print(f"下载连接关闭: {self.client_address}用户: {u}已从挂载点 {mount} 断开")

    def _remove_client(self, client):
        """移除客户端并关闭socket"""
        try:
            client['socket'].shutdown(socket.SHUT_RDWR)
            client['socket'].close()
        except:
            pass
        with clients_lock:
            if client in authenticated_clients:
                authenticated_clients.remove(client)


    # ===================================
    # 挂载点列表生成
    # ===================================
    def _generate_mount_list(self):
        """生成并保存挂载点列表（供客户端查询）"""
        conn = sqlite3.connect('2rtk.db')
        c = conn.cursor()
        try:
            c.execute("SELECT mount, identifier, format, format_details, carrier, nav_system, network, country, latitude, longitude, nmea, solution, generator, compression, authentication, fee, bitrate, misc FROM running_mounts")
            rows = c.fetchall()
            mount_list = []
            for row in rows:
                # 格式: STR;挂载点;其他信息...
                mount_info = ';'.join([str(item) if item is not None else '' for item in ['STR'] + list(row)])
                mount_list.append(mount_info)
            # 添加网络信息
            mount_list.append(f'NET;2RTK;2RTK;N;N;2rtk.com;{HOST}:{NTRIP_PORT};{CONTACT_EMAIL};;')
            mount_list.append('ENDSOURCETABLE;')
            tbl = '\n'.join(mount_list)
            
            # 保存到文件
            with open('mount_list.txt', 'w') as f:
                f.write(tbl)
            logger.info("生成挂载点列表成功")
            print("挂载点列表已更新 mount_list.txt")
        except Exception as e:
            logger.error(f"生成挂载点列表失败: {e}")
        finally:
            conn.close()


# ===================================
# Flask Web管理界面
# ===================================

app = Flask(__name__)
app.secret_key = os.urandom(24)  # 用于会话管理


# 登录验证装饰器
def require_login(func):
    def wrapper(*args, **kwargs):
        if 'admin' not in session:
            next_url = request.url
            return redirect(url_for('login', next=next_url))
        return func(*args, **kwargs)
    wrapper.__name__ = func.__name__
    return wrapper


# 静态文件服务
@app.route('/static/<path:filename>')
def serve_static(filename):
    cache_timeout = 2592000  # 30天缓存
    if filename == 'favicon.ico':
        return send_from_directory(
            os.path.join(app.root_path, 'static'),
            filename,
            mimetype='image/vnd.microsoft.icon',
            cache_timeout=cache_timeout
        )
    return send_from_directory(
        os.path.join(app.root_path, 'static'),
        filename,
        cache_timeout=cache_timeout
    )


# 首页
@app.route('/')
def index():
    # 系统状态
    cpu_percent = psutil.cpu_percent(interval=1)
    mem_percent = psutil.virtual_memory().percent
    start_datetime = datetime.datetime.fromtimestamp(START_TIME)
    formatted_start_time = start_datetime.strftime('%Y-%m-%d %H:%M:%S')
    now = datetime.datetime.now()

    # 运行中挂载点
    running_mounts = []
    with sqlite3.connect('2rtk.db') as conn:
        c = conn.cursor()
        c.execute("SELECT mount, start_time FROM running_mounts")
        running_mounts = [(row[0], datetime.datetime.fromtimestamp(row[1])) for row in c.fetchall()]

    # 在线用户
    running_users = []
    with clients_lock:
        running_users = [
            (
                c['user'], c['mount'], c['agent'], 
                c['addr'][0], c['addr'][1],
                datetime.datetime.fromtimestamp(c['last_sent_ts'])
            ) for c in authenticated_clients
        ]

    return render_template(
        'index.html',
        cpu_percent=cpu_percent,
        mem_percent=mem_percent,
        program_runtime=formatted_start_time,
        running_mounts=running_mounts,
        running_users=running_users,
        now=now,
        VERSION=VERSION,
        CONTACT_EMAIL=CONTACT_EMAIL,
        NTRIP_PORT=NTRIP_PORT
    )


# 登录/登出
@app.route('/login', methods=['GET', 'POST'])
def login():
    # 登出处理
    if request.args.get('logout'):
        session.pop('admin', None)
        return redirect(url_for('index'))

    next_url = request.args.get('next') or request.referrer
    if next_url and url_for('login') in next_url:
        next_url = None  # 避免登录页循环跳转

    # 登录处理
    if request.method == 'POST':
        username = request.form.get('username')
        password = request.form.get('password')
        if not username or not password:
            return render_template('login.html', error='用户名和密码不能为空。')

        with sqlite3.connect('2rtk.db') as conn:
            c = conn.cursor()
            c.execute("SELECT * FROM admins WHERE username =? AND password =?", (username, password))
            if c.fetchone():
                session['admin'] = username
                return redirect(next_url or url_for('index'))
            else:
                return render_template('login.html', error='用户名或密码错误')

    return render_template('login.html', next=next_url)


# 用户管理
@app.route('/user_management', methods=['GET', 'POST'])
@require_login
def user_management():
    error = None
    if request.method == 'POST':
        with sqlite3.connect('2rtk.db') as conn:
            c = conn.cursor()
            # 添加用户
            if 'add_user' in request.form:
                username = request.form.get('username')
                password = request.form.get('password')
                if not username or not password:
                    error = '用户名和密码不能为空。'
                else:
                    c.execute("SELECT * FROM users WHERE username =?", (username,))
                    if c.fetchone():
                        error = '用户名已存在，请选择其他用户名。'
                    else:
                        c.execute("INSERT INTO users (username, password) VALUES (?,?)", (username, password))
                        conn.commit()
                        return redirect(url_for('user_management'))
            # 删除用户
            elif 'delete_user' in request.form:
                user_id = request.form.get('delete_user')
                c.execute("SELECT username FROM users WHERE id =?", (user_id,))
                username = c.fetchone()[0]
                c.execute("DELETE FROM users WHERE id =?", (user_id,))
                conn.commit()
                # 强制该用户的所有连接下线
                with clients_lock:
                    to_remove = [c for c in authenticated_clients if c['user'] == username]
                    for c in to_remove:
                        authenticated_clients.remove(c)
                return redirect(url_for('user_management'))
            # 更新用户
            elif 'update_user' in request.form:
                user_id = request.form.get('update_user')
                username = request.form.get('username')
                password = request.form.get('password')
                if not username or not password:
                    error = '用户名和密码不能为空。'
                else:
                    c.execute("SELECT * FROM users WHERE username =? AND id !=?", (username, user_id))
                    if c.fetchone():
                        error = '用户名已存在，请选择其他用户名。'
                    else:
                        old_username = c.execute("SELECT username FROM users WHERE id=?", (user_id,)).fetchone()[0]
                        c.execute("UPDATE users SET username =?, password =? WHERE id =?", (username, password, user_id))
                        conn.commit()
                        # 强制该用户的所有连接下线
                        with clients_lock:
                            to_remove = [c for c in authenticated_clients if c['user'] == old_username]
                            for c in to_remove:
                                authenticated_clients.remove(c)
                        return redirect(url_for('user_management'))

    # 获取用户列表及在线状态
    with sqlite3.connect('2rtk.db') as conn:
        c = conn.cursor()
        c.execute("SELECT id, username, password FROM users")
        users = c.fetchall()

    with clients_lock:
        online_usernames = {c['user'] for c in authenticated_clients}
    users_with_status = [
        (uid, uname, pwd, "在线" if uname in online_usernames else "离线")
        for uid, uname, pwd in users
    ]

    return render_template(
        'user_management.html',
        users=users_with_status,
        VERSION=VERSION,
        CONTACT_EMAIL=CONTACT_EMAIL,
        error=error
    )


# 挂载点管理
@app.route('/mount_management', methods=['GET', 'POST'])
@require_login
def mount_management():
    error = None
    if request.method == 'POST':
        with sqlite3.connect('2rtk.db') as conn:
            c = conn.cursor()
            # 添加挂载点
            if 'add_mount' in request.form:
                mount = request.form.get('mount')
                password = request.form.get('password')
                if not mount or not password:
                    error = '挂载点名称和密码不能为空。'
                else:
                    c.execute("SELECT * FROM mounts WHERE mount =?", (mount,))
                    if c.fetchone():
                        error = '挂载点名称已存在，请选择其他名称。'
                    else:
                        c.execute("INSERT INTO mounts (mount, password) VALUES (?,?)", (mount, password))
                        conn.commit()
                        return redirect(url_for('mount_management'))
            # 删除挂载点
            elif 'delete_mount' in request.form:
                mount_id = request.form.get('delete_mount')
                c.execute("SELECT mount FROM mounts WHERE id =?", (mount_id,))
                mount = c.fetchone()[0]
                c.execute("DELETE FROM mounts WHERE id =?", (mount_id,))
                conn.commit()
                # 强制该挂载点的所有连接下线
                with clients_lock:
                    to_remove = [c for c in authenticated_clients if c['mount'] == mount]
                    for c in to_remove:
                        authenticated_clients.remove(c)
                # 清理对应RingBuffer缓存
                with rtcm_lock:
                    if mount in rtcm_buffers:
                        del rtcm_buffers[mount]
                return redirect(url_for('mount_management'))
            # 更新挂载点
            elif 'update_mount' in request.form:
                mount_id = request.form.get('update_mount')
                mount = request.form.get('mount')
                password = request.form.get('password')
                if not mount or not password:
                    error = '挂载点名称和密码不能为空。'
                else:
                    c.execute("SELECT * FROM mounts WHERE mount =? AND id !=?", (mount, mount_id))
                    if c.fetchone():
                        error = '挂载点名称已存在，请选择其他名称。'
                    else:
                        old_mount = c.execute("SELECT mount FROM mounts WHERE id=?", (mount_id,)).fetchone()[0]
                        c.execute("UPDATE mounts SET mount =?, password =? WHERE id =?", (mount, password, mount_id))
                        conn.commit()
                        # 强制该挂载点的所有连接下线
                        with clients_lock:
                            to_remove = [c for c in authenticated_clients if c['mount'] == old_mount]
                            for c in to_remove:
                                authenticated_clients.remove(c)
                        # 迁移RingBuffer缓存
                        with rtcm_lock:
                            if old_mount in rtcm_buffers:
                                rtcm_buffers[mount] = rtcm_buffers[old_mount]
                                del rtcm_buffers[old_mount]
                        return redirect(url_for('mount_management'))

    # 获取挂载点列表及运行状态
    with sqlite3.connect('2rtk.db') as conn:
        c = conn.cursor()
        c.execute("SELECT id, mount, password FROM mounts")
        mounts = c.fetchall()
        c.execute("SELECT mount FROM running_mounts")
        running_mounts = {row[0] for row in c.fetchall()}

    return render_template(
        'mount_management.html',
        mounts=mounts,
        running_mounts=running_mounts,
        VERSION=VERSION,
        CONTACT_EMAIL=CONTACT_EMAIL,
        error=error
    )


# 管理员密码修改
@app.route('/change_admin_password', methods=['GET', 'POST'])
@require_login
def change_admin_password():
    if request.method == 'POST':
        old_password = request.form.get('old_password')
        new_password = request.form.get('new_password')
        confirm_password = request.form.get('confirm_password')
        
        if not old_password or not new_password or not confirm_password:
            return "旧密码、新密码和确认新密码不能为空。"
        if new_password != confirm_password:
            return "新密码和确认新密码不一致。"

        with sqlite3.connect('2rtk.db') as conn:
            c = conn.cursor()
            c.execute("SELECT password FROM admins WHERE id = 1")
            result = c.fetchone()
            if result and result[0] == old_password:
                c.execute("UPDATE admins SET password =? WHERE id = 1", (new_password,))
                conn.commit()
                session.pop('admin', None)  # 强制重新登录
                return redirect(url_for('login'))
            else:
                return "旧密码错误"
    
    return render_template('change_admin_password.html', VERSION=VERSION, CONTACT_EMAIL=CONTACT_EMAIL)


# 支付二维码
@app.route('/alipay_qr')
def get_alipay_qr():
    return redirect(ALIPAY_QR_URL)

@app.route('/wechat_qr')
def get_wechat_qr():
    return redirect(WECHAT_QR_URL)


# ===================================
# 服务启动与信号处理
# ===================================
def shutdown_handler(signal, frame):
    """统一信号处理：优雅关闭所有服务"""
    global broadcast_thread_running
    print("\n接收到退出信号，正在清理资源...")
    broadcast_thread_running = False  # 停止广播线程
    
    # 关闭所有客户端socket
    with clients_lock:
        for client in authenticated_clients:
            try:
                client['socket'].shutdown(socket.SHUT_RDWR)
                client['socket'].close()
            except:
                pass
    
    # 关闭NTRIP服务器
    if 'server' in globals():
        server.shutdown()
        logger.info("NTRIP服务器已停止")
    
    print("所有服务已关闭，程序退出")
    sys.exit(0)


# ===================================
# 主程序入口
# ===================================
def main():
    # 1. 初始化数据库
    init_db()
    logger.info("数据库初始化完成")
    
    # 2. 启动广播线程
    global broadcast_thread
    broadcast_thread = BroadcastThread()
    broadcast_thread.daemon = True
    broadcast_thread.start()
    logger.info("广播线程已启动")
    
    # 3. 显示横幅信息
    clear_banner()
    
    # 4. 启动Web管理界面（独立线程）
    web_thread = threading.Thread(
        target=app.run,
        kwargs={
            'host': HOST, 
            'port': WEB_PORT, 
            'use_reloader': False,  # 关闭自动重载（避免多进程）
            'threaded': True       # 启用多线程处理Web请求
        }
    )
    web_thread.daemon = True
    web_thread.start()
    print(f"2RTK Caster Web 管理界面已启动: {HOST}:{WEB_PORT}")
    logger.info(f"Web管理界面启动于 {HOST}:{WEB_PORT}")
    
    # 5. 启动NTRIP服务器
    global server
    try:
        socketserver.ThreadingTCPServer.allow_reuse_address = True
        server = socketserver.ThreadingTCPServer((HOST, NTRIP_PORT), Handler)
    except OSError as e:
        logger.error(f"NTRIP端口绑定失败: {e}")
        print(f"错误: 无法绑定到端口 {NTRIP_PORT} - {e}")
        shutdown_handler(None, None)  # 触发关闭流程
        return
    
    print(f"2RTK NTRIP Caster {VERSION} 已启动: {HOST}:{NTRIP_PORT}")
    logger.info(f"NTRIP服务器启动于 {HOST}:{NTRIP_PORT}")
    
    # 6. 注册信号处理（捕获Ctrl+C等退出信号）
    signal.signal(signal.SIGINT, shutdown_handler)
    signal.signal(signal.SIGTERM, shutdown_handler)
    print("按 Ctrl+C 停止服务")
    
    # 7. 启动服务器主循环
    try:
        server.serve_forever()
    except Exception as e:
        logger.error(f"服务器主循环异常: {e}")
        shutdown_handler(None, None)
    finally:
        # 确保广播线程退出
        broadcast_thread.join(timeout=5)
        if broadcast_thread.is_alive():
            logger.warning("广播线程未能正常退出")


if __name__ == '__main__':
    main()